package com.hubiao.pro.service.mq;

import com.alibaba.fastjson.JSON;
import com.hubiao.pro.common.constant.Constants;
import com.hubiao.pro.common.redis.Cache;
import com.hubiao.pro.common.utils.StringUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.io.*;
import java.util.HashMap;
import java.util.Map;

/**
 * Description
 *
 * @author hubiao
 * @since 2021-01-30 16:29
 */
@Slf4j
@Component
public class DataCenterProducerMq implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {

    /**
     * 重发信息文本存储地址
     */
    private static final String PATH = "C:\\YunChuang\\mqMessage\\\\{}.txt";


    /**
     * 由于rabbitTemplate的scope属性设置为ConfigurableBeanFactory.SCOPE_PROTOTYPE，所以不能自动注入
     */
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Resource
    private Cache cache;

    @PostConstruct
    public void init() {
        //指定 ConfirmCallback
        rabbitTemplate.setConfirmCallback( this );
        //指定 ReturnCallback
        rabbitTemplate.setReturnCallback( this );

    }

    /**
     * 发送到消息队列
     *
     * @param o
     */
    public void send(Object o, String exchange, String routingKey) {
        String uuid = StringUtils.getUUID();
        CorrelationData correlationId = new CorrelationData( uuid );
        Map<Object, Object> headers = new HashMap<>();
        headers.put( "returnMessageNum", 1 );
        headers.put( "returnSendNum", 1 );
        headers.put( "body", JSON.toJSONString( o ) );
        headers.put( "exchange", exchange );
        headers.put( "routingKey", routingKey );
        Message message = MessageBuilder
                .withBody( JSON.toJSONString( o ).getBytes() )
                .setContentType( MessageProperties.CONTENT_TYPE_JSON )
                .setContentEncoding( "utf-8" )
                .setMessageId( uuid )
                .build();
        //设置redis幂等处理的key 防止重复消费
        cache.setnx( StringUtils.format( Constants.MQ_KEY, uuid ), headers.toString() );
        //发送消息
        rabbitTemplate.convertAndSend( exchange, routingKey, message, correlationId );
    }


    /**
     * 回调
     *
     * @param correlationData
     * @param ack
     * @param cause
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        log.info( " 回调id:" + correlationData );
        if (ack) {
            log.info( "消息成功消费" );
        } else {
            String key = StringUtils.format( Constants.MQ_KEY, correlationData.getId() );
            Map<Object, Object> map = JSON.parseObject( cache.get( key ), Map.class );
            int returnSendNum = (int) map.get( "returnSendNum" );
            // 重发三次
            if (returnSendNum > 3) {
                String errorMessage = "return--confirm:" + map.get( "body" ).toString() +
                        ",exchange:" + map.get( "exchange" ).toString() + ",routingKey:" +
                        map.get( "routingKey" ).toString();
                log.error( errorMessage );
                //重发失败超过三次则会写入文件中
                writeIO( errorMessage );
                //删除key
                cache.del( key );
                return;
            }
            //重新封装发送实体
            Message message = MessageBuilder
                    .withBody( map.get( "body" ).toString().getBytes() )
                    .setContentType( MessageProperties.CONTENT_TYPE_JSON )
                    .setContentEncoding( "utf-8" )
                    .setMessageId( correlationData.getId() )
                    .build();
            log.error( "消息消费失败:" + cause );
            map.put( "returnSendNum", ++returnSendNum );
            cache.setnx( key, map.toString() );
            rabbitTemplate.convertAndSend( map.get( "exchange" ).toString(), map.get( "routingKey" ).toString(), message, correlationData );
        }
    }

    /**
     * 消息发送失败回传
     *
     * @param message
     * @param replyCode
     * @param replyText
     * @param exchange
     * @param routingKey
     */
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        String errorMessage = "return--message:" + new String( message.getBody() ) + ",replyCode:" + replyCode + ",replyText:"
                + replyText + ",exchange:" + exchange + ",routingKey:" + routingKey;
        log.error( errorMessage );
        try {
            String key = StringUtils.format( Constants.MQ_KEY, message.getMessageProperties().getMessageId() );
            Map<Object, Object> map = JSON.parseObject( cache.get( key ), Map.class );
            int returnMessageNum = (int) map.get( "returnMessageNum" );
            // 重发两次
            if (returnMessageNum > 3) {
                writeIO( errorMessage );
                cache.del( key );
                return;
            }

            /**
             * TODO 重新发送消息至队列,此处应写一套重发机制,重发多少次结束,否则如果消息如果一直发送失败,则会一直发下去!
             */
            CorrelationData correlationId = new CorrelationData( message.getMessageProperties().getMessageId() );
            map.put( "returnMessageNum", ++returnMessageNum );
            cache.setnx( key, map.toString() );
            rabbitTemplate.convertAndSend( map.get( "exchange" ).toString(), map.get( "routingKey" ).toString(), message, correlationId );
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    /**
     * 写入重发信息到文件
     *
     * @param errorMessage
     */
    private void writeIO(String errorMessage) {
        long currentTime = System.currentTimeMillis();

        File file = new File( StringUtils.format( PATH, currentTime ) );
        if (!file.getParentFile().exists()) {
            file.getParentFile().mkdirs();
        }
        FileWriter fileWriter = null;
        BufferedWriter out = null;
        try {
            if (!file.exists()) {
                file.createNewFile();
            }

            fileWriter = new FileWriter( file );
            out = new BufferedWriter( fileWriter );
            out.write( errorMessage );
            out.flush(); // 把缓存区内容压入文件

        } catch (FileNotFoundException e) {

            log.error( "文件流错误1,error:{}", e.getMessage() );
        } catch (IOException e) {
            log.error( "文件流错误2,error:{}", e.getMessage() );
        } finally {
            try {
                out.close();
                fileWriter.close();
            } catch (IOException e) {
                log.error( "文件流错误3,error:{}", e.getMessage() );
            }

        }
    }
}
